package com.atguigu.app

import com.alibaba.fastjson.JSON
import com.atguigu.gmall.constances.CommonConstants
import com.atguigu.util.MykafkaUtil.MyKafkaUtil
import com.atguigu.util.PropertiesUtil
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

import java.util.Properties

object UserInfoApp extends BaseApp {
  override var appName: String = "UserInfoApp"
  override var internal: Int = 10

  def main(args: Array[String]): Unit = {
    sc=new StreamingContext("local[4]",appName,Seconds(internal))
    runApp{

      val ds = MyKafkaUtil.getKafkaStream(CommonConstants.KAFKA_USER_INFO, sc)
      ds.foreachRDD(rdd=>{
        rdd.foreachPartition(partition=>{

          val config: Properties = PropertiesUtil.load("config.properties")
          val client: Jedis = new Jedis(config.getProperty("redis.host"), config.getProperty("redis.port").toInt)
          partition.foreach(line=>{

            val jsonstr = JSON.parseObject(line.value())

            println(jsonstr)

            client.set("user_info:"+jsonstr.get("id"),line.value())

          })

          client.close()

        })
      })





    }
  }
}
